#pragma once
#include <thread>
#include <unordered_map>
#include "function_traits.h"
#include "caller.h"
#include "localstream_asio.h"

namespace KRPC {

template<typename ... Args>
class SubcribeByAsio
{	
public:
	SubcribeByAsio(std::string topic, LOCALSTREAM_ASIO &_com, uint32_t timeoutMsec) : \
		com(_com), name(std::move(topic)), msec(timeoutMsec) 
	{
		assert(name.size() < MAX_RPCNAME_LEN);
	}
	
	void start(std::function<void(Args...)> deal)
	{
		com.start(
			[&](Packet &packet, const SendFunc &sndfunc)->bool{
				bool ret = packet.valid();
				if (packet.valid()) {
					sndfunc(package());	//收到消息时立即回复一次避免服务端超时关闭订阅的连接
					call(deal, packet);
				} 
				return ret;
			},
			msec,
			[&](const SendFunc &sndfunc){
				sndfunc(package());
			}
		);
	}
private:
	LOCALSTREAM_ASIO &com;
	std::string name;
	uint32_t msec;	//接收超时秒数

	Packet package()
	{
		Packet packet(sizeof(MsgHead), sizeof(MsgHead));
		MsgHead *head = packet.getHead();
		std::memset(head, 0, sizeof(MsgHead));
		head->mode = RPC_SUB;
		//head->ver = com.isLocal() ? 0 : 1;	//非本地连接则默认使用有校验版本,本地连接则默认使用无校验版本
		//head->id = ++id;
		std::memcpy(head->name, name.c_str(), name.size());
		head->name[name.size()] = '\0';
		packet.pack();
		return std::move(packet);
	}
};

template<>
class SubcribeByAsio<void> 
{	//禁止订阅void类型数据
};

class MultiSubcribeByAsio
{	
public:
	MultiSubcribeByAsio(LOCALSTREAM_ASIO &_com, uint32_t timeoutMsec) : com(_com), msec(timeoutMsec) {}

	template<typename F>
	void subcribe(std::string topic, F &&func)
	{
		assert(functions.find(topic) == functions.end());
		functions.emplace(std::move(topic), std::bind(Caller<decay_t<F>>(), 
			wrapper(std::forward<F>(func)), std::placeholders::_1));
	}

	template<typename F, typename Self>
	void subcribe(std::string topic, F &&func, Self *self)
	{
		assert(functions.find(topic) == functions.end());
		functions.emplace(std::move(topic), std::bind(Caller<decay_t<F>, Self>(), 
			wrapper(std::forward<F>(func)), self, std::placeholders::_1));
	}

	void start()
	{
		if (functions.empty())
			return;
		com.start(
			[this](Packet &packet, const SendFunc &sndfunc)->bool{
				bool ret = packet.valid();
				if (packet.valid()) {
					const MsgHead *h = packet.getHead();
					auto it = functions.find(h->name);
					if (h->mode == RPC_PUB && it != functions.end() && it->second(packet)) {
						sndfunc(package(it->first));
					} /*else {	//订阅回复的所有失败信息包括无权限,订阅类型不匹配等都默认删除订阅回调函数
						if (it != functions.end())
							functions.erase(it);
					} */ 
				}
				return ret;
			},
			msec,
			[this](const SendFunc &sndfunc){
				for (auto &it : functions) {
					sndfunc(package(it.first));
				}
			}
		);
	}

private:
	LOCALSTREAM_ASIO &com;
	uint32_t msec;	//接收超时秒数
	std::unordered_map<std::string, std::function<bool(Packet &)>> functions;

	Packet package(const std::string &name)
	{
		Packet packet(sizeof(MsgHead), sizeof(MsgHead));
		MsgHead *head = packet.getHead();
		std::memset(head, 0, sizeof(MsgHead));
		head->mode = RPC_SUB;
		//head->ver = com.isLocal() ? 0 : 1;	//非本地连接则默认使用有校验版本,本地连接则默认使用无校验版本
		//head->id = ++id;
		std::memcpy(head->name, name.c_str(), name.size());
		head->name[name.size()] = '\0';
		packet.pack();
		return std::move(packet);
	}
};

} //end of namespace KRPC
